const { parentPort } = require('worker_threads')
const WebSocketServer = require('ws').Server;
const http = require('http');
const fs = require('fs-extra');
const dayjs = require('dayjs');
const { omit } = require('lodash');
const child_process = require('child_process')
const uuid = require('uuid')

const previewData = {}
const pushVideoTime = {}

const defaultOptions = {
  path: '/ws',
  skipUTF8Validation: true,
}

const WS_STATUS = {
  CONNECTING: 0,
  OPEN: 1,
  CLOSING: 2,
  CLOSED: 3
};

let wsLogs = '/civi/ws_logs.log'
if (fs.existsSync()) fs.unlinkSync(wsLogs)
// let wsLogSteam = fs.createWriteStream(wsLogs)

var
  WSS,
  WSS_CLIENTS = {}
WSS_RTSP = {};

var PLAYER_YUV = {};
var previewPipe = {}
var clientRecord = {}



class PreviewPipe {
  constructor(rtsp) {
    this.rtsp = rtsp
    this.pipe = '/civi/pipe/' + md5(rtsp)
    // 创建管道文件夹
    fs.ensureDirSync('/civi/pipe')
    // 创建管道文件
    createPipe(this.pipe)

    this.readStream = fs.createReadStream(this.pipe)
    this.writeStream = fs.createWriteStream(this.pipe)

    // 读取管道数据
    let pipeData = ''
    this.readStream.on('data', (data) => {
      pipeData += data.toString()
      if (pipeData.indexOf('##end') > -1) {
        try {
          let data = JSON.parse(pipeData.split('##end')[0])
          previewData[this.rtsp] = { img: data.data.img, time: Date.now(), otime: data.data.time, rtsp: this.rtsp, size: data.data.size }
        } catch (e) {
          console.error('格式错误', e)
        }
        pipeData = ''
      }
    })
  }
}

function createPipe(name) {
  if (fs.existsSync(name)) return;
  try {
    let result = child_process.execSync(`mkfifo ${name}`)
  } catch (e) {
    console.error('创建失败', e)
  }
}

function md5(str) {
  let data = child_process.execSync(`echo "${str}" | openssl md5`)
  return data.toString().split(' ')[1].trim()
}

/**
 * 创建WSS服务
 * @param {*} server 
 * @param {*} config 
 */
function createWss(server, config) {
  config = Object.assign(
    { server }, defaultOptions,
    omit(config || {
      perMessageDeflate: true
    }, ['host', 'port', 'backlog', 'server', 'verifyClient', 'handleProtocols', 'noServer'])
  )

  // WSS = new WebSocketServer(config)
  WSS = new WebSocketServer(config)
  // 监听错误
  WSS.on('error', (err) => {
    parentPort.postMessage({ event: 'error', data: err })
  })

  // 监听连接事件
  WSS.on('connection', (ws) => {
    
    // if (ws._socket.remoteAddress == '192.168.18.68') {
    //   ws.binaryType = 'arraybuffer'
    //   console.log('收到连接', ws._socket.remoteAddress)
    // }
    // return
    // 收到客户端连接 响应连接成功信号
    ws.send(JSON.stringify({ event: 'open', data: 'success' }))
    // 监听客户端发送消息
    ws.on('message', (message) => {
      // 格式化收到消息
      let data;

      // 通过 /JSON 分割 前面是JSON后面是图片
      let index = message.indexOf(Buffer.from('##JSON_END##'))
      if (index > -1) {
        // JSON数据
        let json = message.slice(0, index)
        // 图片数据
        let img = message.slice(index + 5)

        // 格式化JSON
        data = JSON.parse(json.toString())

        // 如果img长度大于100 则认为是图片数据
        if (img.length > 100) {
          data.data.img = img
        }
      } else {
        try {
          data = JSON.parse(message.toString())
        } catch (err) { parentPort.postMessage({ event: 'error', data: new Error('Expect to receive a JSON string') }) }
      }

      if (!data) return

      // 数据不符合规范
      if (!data.event) {
        parentPort.postMessage({ event: 'error', data: new Error('The received data format does not match. Please refer to `{"event": "event", "data": "data"}`') })
      }

      // 前端响应连接
      if (data.event === 'clientOpen') {
        let uuid = data.data
        if (WSS_CLIENTS[uuid]) {
          WSS_CLIENTS[uuid].close()
          delete WSS_CLIENTS[uuid];
        }
        ws.uuid = uuid;
        WSS_CLIENTS[uuid] = ws;
      }
      // 指定接收的数据
      else if (data.event === 'rtspDisplay') {
        let uuid = data.data.uuid, rtsp = data.data.rtsp
        ws.uuid = uuid;
        ws.rtsp = rtsp
        // 获取客户端的IP
        let ip = ws._socket.remoteAddress
        // console.log(ip)

        if (WSS_RTSP[uuid]) {
          WSS_RTSP[uuid].close();
        }
        WSS_RTSP[uuid] = ws;
        parentPort.postMessage({ event: 'rtspDisplay', data: rtsp })
        // // 开启管道
        // if (!previewPipe[rtsp]) {
        //   previewPipe[rtsp] = new PreviewPipe(rtsp)
        // }
      }
      // 前端指定当前连接接收的相机数据
      else if (data.event === 'cameraConnection') {
        ws.camera = data.data

      }
      // YUV预览
      else if (data.event === 'yuvPlayer') {
        let uuid = data.data.uuid, rtsp = data.data.rtsp
        ws.uuid = uuid;
        ws.rtsp = rtsp
        if (PLAYER_YUV[uuid]) {
          parentPort.postMessage({ event: 'yuvPlayerClose', data: PLAYER_YUV[uuid].rtsp })
          PLAYER_YUV[uuid].close();
        }
        PLAYER_YUV[uuid] = ws;
        parentPort.postMessage({ event: 'yuvPlayerOpen', data: rtsp })
      }

      else if (data.event === 'get_preview_data') {
        let rtsp = data.data.rtsp
        if (rtsp) {
          let _data = previewData[rtsp]
          if (_data) {
            try {
              if (typeof _data.img === 'string') {
                ws.send(JSON.stringify({ event: 'preview_data', _data }))
              } else {
                let json = JSON.stringify({ event: 'preview_data', time: _data.time, rtsp: _data.rtsp, size: _data.size })
                let buf = Buffer.from(json)
                buf = Buffer.concat([buf, Buffer.from('/JSON'), _data.img])
                ws.send(buf)
              }
            } catch (e) {
              console.log('发送数据报错', e)
            }
          }
        }
      }
      else if (data.event === 'pushVideo') {
        // wsLogSteam.write(`【${dayjs().format('YYYY-MM-DD HH:mm:ss.SSS')}】-[${data.data.type}]-${data.data.cam}\n`)
        // 修复接收频率过高导致内存泄漏
        let rtsp = data.data.cam
        if (data.data.type === 'preview') {
          previewData[rtsp] = { img: data.data.img, time: Date.now(), rtsp, size: data.data.size }
          try { ws.send("OK") } catch (e) { console.error(e) }
        } else {
          if (data.data.type == 'intrusion_detection') {
            // console.log('收到算法推送数据', rtsp)
          }
          pushVideoTime[rtsp] = Date.now()
          parentPort.postMessage(data)
        }
      }

      // 其他事件 发送到主线程处理
      else {
        // if (data.event === 'pushVideo') broadcastRtsp(data.data)
        parentPort.postMessage(data)
      }
    })
    // 错误监听
    ws.on('error', function (err) {
      console.log('处理出错', err)
    });
    // 监听客户端关闭事件
    ws.on('close', () => {
      if (ws.uuid && WSS_CLIENTS[ws.uuid]) {
        WSS_CLIENTS[ws.uuid].close();
        delete WSS_CLIENTS[ws.uuid];
      } else if (ws.uuid && WSS_RTSP[ws.uuid]) {
        let rtsp = WSS_RTSP[ws.uuid].rtsp
        WSS_RTSP[ws.uuid].close();
        delete WSS_RTSP[ws.uuid];

        // 循环是否还有当前RTSP的预览连接
        let _display = Object.keys(WSS_RTSP).find(uuid => WSS_RTSP[uuid].rtsp == rtsp)
        if (!_display) {
          // 没有发送关闭消息
          parentPort.postMessage({ event: 'rtspDisplayClose', data: rtsp })
        }
      } else if (ws.uuid && PLAYER_YUV[ws.uuid]) {
        let rtsp = ws.rtsp
        ws.close();
        delete PLAYER_YUV[ws.uuid];
        parentPort.postMessage({ event: 'yuvPlayerClose', data: rtsp })
      } else {
        ws.close();
      }
    })
  })

  // 通知主线程创建成功
  parentPort.postMessage({ event: 'create_ok' })
}


function broadcastRtsp(data) {
  // let img = Buffer.from(data.img, 'base64'), type = data.type, rtsp = data.cam
  let img, type = data.type, rtsp = data.cam

  // 分割buffer
  // let list = [], len = 100
  // for (let i = 0; i < img.length / len; i++) {
  //   let start = i * len, end = (i + 1) * len;
  //   if (end >= img.length - 2) {
  //     end = img.length - 2;
  //     list.push(img.slice(start, end))
  //     break;
  //   } else {
  //     list.push(img.slice(start, end))
  //   }
  // }
  // list.push(img.slice(img.length - 2, img.length))


  Object.keys(WSS_RTSP).forEach(uuid => {
    if (WSS_RTSP[uuid].display.type === type && WSS_RTSP[uuid].display.rtsp === rtsp) {
      // list.forEach(buf => WSS_RTSP[uuid].send(buf))
      let client = WSS_RTSP[uuid];
      if (client.bufferedAmount <= 0 && client.readyState === WS_STATUS.OPEN) {
        client.send(data.img)
      }
    }
  })
}


function broadcast(event, data, wait = true) {
  let buf = JSON.stringify({ event, data })
  // if (event == 'message_change') console.log('触发消息发送', Object.keys(WSS_CLIENTS).length)
  Object.keys(WSS_CLIENTS).forEach(uuid => {
    let client = WSS_CLIENTS[uuid]
    // wait 等于false 并且客户端数据未接收完成 返回
    if (!wait && client.bufferedAmount > 0) return;
    // 客户端状态打开 发送消息
    if (client.readyState === WS_STATUS.OPEN) {
      client.send(buf)
      // if (event == 'message_change') console.log('发送消息', data._id)
    }
  })
}

function broadcastPull(data) {
  Object.keys(WSS_CLIENTS).forEach(uuid => {
    let client = WSS_CLIENTS[uuid]
    if (client.bufferedAmount > 0) return;
    if (client.readyState === WS_STATUS.OPEN && client.camera) {
      try {
        let text = JSON.stringify({
          event: 'pullVideo',
          data: {
            [client.camera.type]: {
              [client.camera.rtsp]: data[client.camera.type][client.camera.rtsp]
            }
          }
        })
        client.send(Buffer.from(text, 'utf-8'))
      } catch (e) { }
    }
  })
}


function broadcastDisplay(imgdata) {
  let { rtsp, data } = imgdata
  Object.keys(WSS_RTSP).forEach(uuid => {
    let client = WSS_RTSP[uuid]
    if (client.rtsp == rtsp) {
      if (client.bufferedAmount <= 0 && client.readyState === WS_STATUS.OPEN) {
        client.send(data)
      }
    }
  })
}

function broadcastYUV(data) {
  let { rtsp, yuv } = data
  Object.keys(PLAYER_YUV).forEach(uuid => {
    let client = PLAYER_YUV[uuid]
    if (client.rtsp == rtsp) {
      if (client.bufferedAmount <= 0 && client.readyState === WS_STATUS.OPEN) {
        client.send(yuv)
      }
    }
  })
}

parentPort.on('message', (data) => {
  // 创建事件
  if (data.event === 'create') {
    let config = data.data
    let server = http.createServer();
    server.listen(config.port, '0.0.0.0', () => {
      createWss(server, config)
    })
  }
  // 广播事件
  else if (data.event === 'broadcast') {
    // 提取数据
    let { event, wait } = data.data
    let _data = data.data.data
    broadcast(event, _data, !!wait)
  }
  // 广播pull事件
  else if (data.event === 'broadcastPull') {
    broadcastPull(data.data)
  }
  // 广播pull事件
  else if (data.event === 'broadcastDisplay') {
    broadcastDisplay(data.data)
  }
  // 广播YUV事件
  else if (data.event === 'broadcastYUV') {
    broadcastYUV(data.data)
  }
  else if (data.event === 'preview_rtsp_data') {
    Object.keys(data.data).forEach(rtsp => {
      previewData[rtsp] = data.data[rtsp]
    })
  }
})